package com.dongdongshop.mq;

import com.dongdongshop.constant.AlipayConstant;
import com.dongdongshop.pojo.TbOrder;
import com.dongdongshop.pojo.TbPayLog;
import com.dongdongshop.service.OrderService;
import com.dongdongshop.service.PayLogService;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;
import java.util.Map;

/**
 * @Description
 * @ClassName RocketMQLocalListener
 * @Author RC
 * @date 2020.12.27 14:36
 */
// txProducerGroup : 要和发送的事务组一致
@RocketMQTransactionListener(txProducerGroup = "tx-group-order")
public class RocketMQLocalListener implements RocketMQLocalTransactionListener {
    @Resource
    private RedisTemplate redisTemplate;
    @Resource
    private OrderService orderService;
    @Autowired
    private PayLogService payLogService;
    private static final Logger logger = (Logger) LoggerFactory.getLogger(RocketMQLocalListener.class);
    /*
     *   MQ服务，成功接到事物消息后，执行本方法；
     *   处理本地事物，并将本地事物处理结果返回给MQ服务
     *
     *   如果我们发送的半消息成功，那么MQ就会回调该方法
     *   那么我们就在当前方法执行业务逻辑（删减库存）
     *   message : 我们发送的消息数据
     *   arg : 我们扣减库存所需要的参数
     * */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object arg) {
        // 将本地事物执行状态return到MQ服务，并且保存到redis备份（网络延迟，MQ服务产生回调的时候 可以从redis中获取）
        try {
            // 业务处理
            /*System.out.println("mq发送的消息：" + new String((byte[]) message.getPayload()));
            System.out.println("操作数据库所需要的参数：" + arg);
            System.out.println("消息ID ：" + message.getHeaders().get("TRANSACTION_ID"));
            System.out.println("事务ID ：" + message.getHeaders().getId());*/

            Map<String,Object> map = (Map<String, Object>) arg;
            String out_trade_no = (String)map.get("out_trade_no");
            String trade_no = (String)map.get("trade_no");
            String totalFee = (String)map.get("totalFee");
            String userId = (String)map.get("userId");
            List<TbOrder> orderList = (List<TbOrder>)map.get("orderList");
            Double d = Double.parseDouble(totalFee);

            String orderIds = "";                   // 拼装订单编号
            for (TbOrder order : orderList) {
                orderIds +=  "," + order.getOrderId();
            }
            orderIds.substring(1);
            TbPayLog payLog = new TbPayLog();
            payLog.setOutTradeNo(out_trade_no);         // 支付订单号
            payLog.setCreateTime(new Date());           // 创建日期
            payLog.setPayTime(new Date());              // 支付完成日期
            payLog.setTotalFee(d.longValue());          // 支付金额
            payLog.setUserId(userId);                   // 用户id
            payLog.setTransactionId(trade_no);          // 交易号码
            payLog.setTradeState(orderList.get(0).getStatus());                  // 交易状态
            payLog.setOrderList(orderIds);              // 订单编号列表

            orderService.updateOrderStatus(out_trade_no);
            payLogService.savePayLogInfo(payLog);
            // 订单支付完成，删除redis中当前用户的购物车数据
            redisTemplate.boundHashOps("cartList").delete(userId);
            redisTemplate.delete(AlipayConstant.TRADE_ID + userId);

            // int i = 9 / 0;
        } catch (Exception e) {
            // 处理异常返回ROLLBACK
            redisTemplate.opsForValue().set(message.getHeaders().get("TRANSACTION_ID"), "2");
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        // 处理成功返回COMMIT
        redisTemplate.opsForValue().set(message.getHeaders().get("TRANSACTION_ID"), "1");
        return RocketMQLocalTransactionState.COMMIT;
    }


    // MQ服务 由于网络等原因 未收到 本地事物处理结果，回查本地事物处理结果
    // 当我们本地事务执行成功或者失败后，提交或者回滚消息的时候，由于网路原因，导致mq一直没有收到消息，mq就会自动回查当前方法
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        //通过事务id获取 redis中 对应的本地事务执行状态
        String status = (String) redisTemplate.opsForValue().get(message.getHeaders().get("TRANSACTION_ID").toString());

        if ("1".equals(status)) {
            return RocketMQLocalTransactionState.COMMIT;
        } else if ("2".equals(status)) {
            // 继续处理，或者直接回滚
            // 处理成功后，可以继续COMMIT，消费者就能正常消费
            // 不处理，直接回滚消息也可以
            return RocketMQLocalTransactionState.ROLLBACK;
        } else {
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }
}
